package com.hopu.user.time;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.time.TimeEra;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

public class TimeEraAna {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<TimeEra, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String hour = json.getJSONObject("odate").getString("hour");
            String era = json.getJSONObject("birthday").getString("year").substring(2, 3) + "0";

            return new Tuple2<>(new TimeEra(hour, era), 1);
        });

        for (int i = 0; i < 24; i++) {
            transAndPut(pairRDD, i+"");
        }
    }

    public static void transAndPut(JavaPairRDD<TimeEra, Integer> pairRDD, String hour) {
        //过滤时间
        pairRDD = pairRDD.filter(t -> t._1.getHour().equals(hour));

        String[] eras = {"50", "60", "70", "80", "90", "00", "10"};
        TimeEra result = new TimeEra(hour, "");
        for (int i = 0; i < eras.length; i++) {
            Integer count = filterEra(pairRDD, eras[i]);
            result.getCounts().add(count);
        }

        //计算某个时间段的所有年代对应的占比
        //得到某个时间段所有操作总量
        int total = 0;
        for (int count: result.getCounts()) {
            total += count;
        }
        List<String> percents = result.getPercents();
        for (int count: result.getCounts()) {
            String percent = (float)count * 100 / total + "%";
            percents.add(percent);
        }

        System.out.println(result);
        //把结果put到HBase当中
        String[] columns = {"hour", "50count", "60count", "70count", "80count", "90count", "00count", "10count",
                            "50percent", "60percent", "70percent", "80percent", "90percent", "00percent", "10percent"};
//        System.out.println(result.getCounts().get(0));
//        System.out.println(result.getCounts().get(1));
//
//        System.out.println(result.getPercents().get(0));
//        System.out.println(result.getPercents().get(1));

        String[] values = {result.getHour(), result.getCounts().get(0)+"", result.getCounts().get(1)+"",
                result.getCounts().get(2)+"", result.getCounts().get(3)+"", result.getCounts().get(4)+"", result.getCounts().get(5)+"",
                result.getCounts().get(6)+"", result.getPercents().get(0), result.getPercents().get(1), result.getPercents().get(2),
                result.getPercents().get(3), result.getPercents().get(4), result.getPercents().get(5), result.getPercents().get(6)};

        try {
            HBaseUtils.putOneRowToHbase("time", result.getHour(), "erainfo", columns, values);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static Integer filterEra(JavaPairRDD<TimeEra, Integer> pairRDD, String era) {
        //得到某个年代的结果
        pairRDD = pairRDD.filter(t -> t._1.getEra().equals(era));
        JavaPairRDD<TimeEra, Iterable<Integer>> groupByHourAndEra = pairRDD.groupByKey();

        JavaRDD<Integer> map = groupByHourAndEra.map(t -> {
            int count = 0;
//            TimeEra result = t._1;
            for (int v : t._2) {
                count += v;
            }
//            result.getCounts().add(count);
            return count;
        });
        Integer first = map.first();
        return first;
    }
}
